package com.ydl.learning.flink.demo

import com.ydl.learning.flink.demo.mySink.LogSink
import java.util.UUID
import org.apache.flink.api.common.state.MapStateDescriptor
import org.apache.flink.api.common.typeinfo.BasicTypeInfo
import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.functions.co.KeyedBroadcastProcessFunction
import org.apache.flink.util.Collector

/**
 *
 *
 * @author ydl
 * @since 2021/4/15
 */
object Test extends Utils {
  def main(args: Array[String]): Unit = {
    val info = getSocketTextStream(9999)
    info._1.setParallelism(1)
    info._1.disableOperatorChaining()
    val value = info._2.map(a => {
      val b = a.split(" ")
      (b(0), b(1))
    }).keyBy(_._1)

    val confStream = streamEnv.readTextFile("/Users/ydl/work/tmp/conf")
    val broadcastStateDescriptor: MapStateDescriptor[String, String] = new MapStateDescriptor[String, String](
      "broadcast",
      BasicTypeInfo.STRING_TYPE_INFO,
      BasicTypeInfo.STRING_TYPE_INFO
    )
    val broadcastConfStream = confStream.broadcast(broadcastStateDescriptor)
    val value1 = value.connect(broadcastConfStream)
    val value2 = value1.process(new KeyedBroadcastProcessFunction[String, (String, String), String, (String, String)] {
      private val descriptor_ = broadcastStateDescriptor

      override def processElement(value: (String, String),
                                  ctx: KeyedBroadcastProcessFunction[String, (String, String), String, (String, String)]#ReadOnlyContext,
                                  out: Collector[(String, String)]): Unit = {
        val str = ctx.getBroadcastState(descriptor_).get(value._1)
        println(str)
        out.collect((str, value._2))
      }

      override def processBroadcastElement(value: String,
                                           ctx: KeyedBroadcastProcessFunction[String, (String, String), String, (String, String)]#Context,
                                           out: Collector[(String, String)]): Unit = {
        val uuid = UUID.randomUUID().toString
        ctx.getBroadcastState(descriptor_).put(value, uuid)
        println(value, uuid)
      }
    })
    value2.addSink(new LogSink[(String, String)])
    info._1.execute("Socket Window WordCount")
  }
}
